package org.example;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

/**
 * @description: 消费者
 * @author: xukang
 * @create: 2025-05-29
 */
public class KafkaConsumerExample4 {
    public static void main(String[] args) {
        // 配置消费者属性
        Properties props = new Properties();
        // ip 端口
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // 消费组
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "signe-group");
        // 反序列化
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 从最早的消息开始消费
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 是否自动提交,true就不需要手动commitSync
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 设置单次 poll 最大返回量
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "500");
        
        // 创建消费者实例
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            // 订阅主题
            consumer.subscribe(Collections.singletonList("my-topic"));
            System.out.println("我是消费组:signe-group");
            // 持续拉取消息
            while (true) {
                // 若有可用消息，立即返回 . 若无消息，等待 100ms 后返回空集合（ConsumerRecords）。
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1));

                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                // 指定分区
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                    // 处理分区数据（示例：只处理前5条）
                    for (int i = 0; i < Math.min(5, partitionRecords.size()); i++) {
                        ConsumerRecord<String, String> record = partitionRecords.get(i);

                        // 记录当前处理到的 Offset（下一次要消费的位置）
                        long offset = record.offset() + 1;
                        offsets.put(partition, new OffsetAndMetadata(offset));
                    }
                }


                // 手动同步提交偏移量,这样就不会重复推送,必须等到offset 提交完毕,再去消费下一批数据.
                //consumer.commitSync();
                // 手动异步提交偏移量,提交完毕,就去消费下一批数据,不用等broker确认
                // consumer.commitAsync();
                // 提交指定 Offset（重点方法）

                consumer.commitSync(offsets);
                // 异步提交指定 Offset
//                consumer.commitAsync(offsets, new OffsetCommitCallback() {
//                    @Override
//                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
//                        if (exception != null) {
//                            log.error("Commit failed for offsets {}", offsets, exception);
//                        }
//                    }
//                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
